package peerToPeerFacebook.proj;


import java.io.IOException;
import java.io.Serializable;
import java.lang.reflect.Method;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import peerToPeerFacebook.proj.DataManager.TransactionStatus;
import peerToPeerFacebook.proj.Log.LogRecord;
import peerToPeerFacebook.proj.ResultGatherer.Result;
import peerToPeerFacebook.src.edu.washington.cs.cse490h.lib.Callback;

/** This class is responsible for coordinating work between all data managers. It is also responsible for restoring
 * them and making sure that they completely initialize after the crash.
 * 
 * For most part this class is network agnostic and works only with local Data Managers. It will also implement 2PC
 * protocol for committing transaction. When client would like to commit transaction it should call into this class.
 */
public class DataManagerCoordinator
{
	/** Set of possible records in the log */
	protected enum LogRecords {
		/** State that indicates that new transaction has been started. 
		 * Params: ClientID, transactionSuffix */
		Started, 
		
		/** Transaction entered pre-commit phase. Due to the way code is written transaction can be either in Running
		 * or pre-commit. Depends on when the crash happened. It is assumed that this record is preceeded with Start
		 * record for this transaction.
		 * Params: trasnactionID, peers[] */ 
		PreCommit, 
		
		/** Marks that certain transaction was completed. This should eitehr be on its own or preceeded by PreCommit.
		 * Params: Transaction ID */
		Commited,
		
		/** Transaction was aborted. Any of the records can preceed this one.
		 * Params: Transaction ID */
		Aborted
	};
	
	/** Node that is used for network operations and file IO */
	protected FBNode node = null;
	
	/** List of active transactions. Key is client ID and value is DataManager for this client */
	protected Map<Integer, DataManager> transactions = new HashMap<Integer, DataManager>(); 
	
	/** List of completed transactions and their statuses */
	protected Map<String, TransactionStatus> completed = new HashMap<String, TransactionStatus>();
	
	/** Manager that distributes locks to different Data Manager. */
	protected LockManager locks = null;
	
	/** Log used by this manager. This log will never be cleaned as we do not know when some may decide to check if
	 * transaction is still alive.
	 */
	protected Log log = null;

	/** Initializes instance of this class. Constructor is responsible for making sure that any already running
	 * transaction is correctly restored.
	 * 
	 * @param node Node used for network and file IO 
	 */
	public DataManagerCoordinator(FBNode node) throws IOException
	{
		this.node = node;
		this.log = new Log(new NodeWrapper(node), "DMC-log");
		
		// It is important that only one instance of this class is used by DMC and all DMs.
		this.locks = new LockManager();	
		loadFromLog();
	}
	

	/**
	 * set the status of the transaction in the completed Map
	 * @param transactionSuffix the ID of the transaction which needs to be set
	 * @param status the status of the transaction
	 */
	public void setTransactionStatus(String transactionSuffix, TransactionStatus status) throws IOException
	{
	    assert status == TransactionStatus.Comitted || status == TransactionStatus.Aborted: "DMC only wants to know " +
	    		"about final outcome of the transaction. Thus only committed and aborted as supported states";

        // Client has no idea about our ID schema. So make sure to convert to real transaction.
        String transactionID = generateTransactionID(node.addr, transactionSuffix);

	    // If we are running on the same node as coordinator, it is possible that we may get that one instead of commit
	    // or abort. This is especially likely if we are restoring after the crash.
	    // However, if we abort or commit transaction here, we will not be able to do that later if commit is called
	    if (transactions.containsKey(node.addr))
	        if (status == TransactionStatus.Comitted)
	            commit(node.addr);
	        else
	            abort(node.addr);
	    else
    	    // We need to update our log about transaction status. As there is no easy way to figure out when clients
    	    // finished querying us, we need to remember that forever. By the time client queries us it is possible that
    	    // TM running this transaction is long gone. So we can't relay on it notified us after the crash
    	    log.writeMark(status == TransactionStatus.Comitted ? 
    	            LogRecords.Commited.toString() : LogRecords.Aborted.toString(), transactionID);
	    
	    // check that if we have transaction it is of the same status
	    assert !completed.containsKey(transactionID) || completed.get(transactionID) == status: "If we already have " +
	    		"transaction record it should be the same as passed in";
	    
	    this.completed.put(transactionID, status);
	}

	/** Starts new transaction for the given client. As only one transaction is supported by client, we do not require
	 * any transaction identification.
	 * 
	 * @param clientID Client making a call - provided by RPC Layer
	 * @param transactionSuffix Unique transaction ID for the client. This does not need to be unique for the network,
	 * 	but client needs to guarantee that this is unique for it. I.e. Client should not try to start transaction with
	 *  same ID twice.
	 */
	public void startTransaction(int clientID, String transactionSuffix) throws IOException
	{
	    // See if we have this transaction running. If we do, we need to abort it and only then start new one. At this
	    // point we assume that client abandoned, thus implicitly aborted transaction
	    if (transactions.containsKey(clientID))
	    {
	        DataManager dm = transactions.get(clientID);
	        System.out.println("Client " + clientID + " decided to start new transaction without finishing old ones. " +
	        		"Aborting this transaction. Old trasnactionID: " + dm.transactionID() + ", status: " + dm.getSatus());
	        abort(clientID);
	    }
	    
		// As we support only one transaction at the time, we do not care about the ID. So make sure it is unique and
		// move on.
		assert !transactions.containsKey(clientID): "Client: " + clientID + " already has an active transaction";
		
		System.out.println("DMC: Started a new transaction with ID: " + transactionSuffix);
		
		// Mark that we are about to run transaction and start it
		log.writeMark(LogRecords.Started.toString(), clientID + " " + transactionSuffix);
		transactions.put(clientID, new DataManager(generateTransactionID(clientID, transactionSuffix),
				new NodeWrapper(node), locks));		
	}

	/** Returns transaction id for given client and suffix provided by the client. As long as suffix is unique for
	 * client, ID returned by this function is unique for the system.
	 * 
	 * @param clientID ID of the client
	 * @param transactionSuffix Suffix provided by client. Has to be unique for that client
	 * @return Generated transaction ID.
	 */
	protected String generateTransactionID(int clientID, String transactionSuffix)
	{
		return clientID + "-" + transactionSuffix;
	}
	
	/** Checks if transaction associated with the client can be committed. This is implemented in such way, that if
	 * vote is yes, it is guaranteed that transaction can be later committed.
	 *   
	 * @param coordinatorID Client making a call - provided by RPC Layer
	 * @param peers List of peers that participated in this transaction. Will be used to check what is outcome of
	 * 	transaction if client fails to tell us that,
	 * @return True if transaction can be committed. 
	 */
	public boolean voteRequest(int coordinatorID, int[] peers) throws IOException
	{
        transactionExistsGuard(coordinatorID);
		DataManager dm = transactions.get(coordinatorID);
		
		// Check if it is good time to commit transaction
		boolean commit = dm.status == TransactionStatus.Running || dm.status == TransactionStatus.PreCommit;		
		assert commit: "If we have transaction it should be always in commitable state";
		
		// Convert IDs to string list
		String str = "";
		for (Integer i: peers)
			str += i + " ";
		str.trim();
		
		// And if it is move it to precommit state so that we could guarantee commit
		// Update log to make sure that we know that it is pre-commited and who are our peers.
		// DM wil be the one that is keeping record if it succesfully completed or not. It is possible that our logs
		// says preCommioted, while DMs says aborted. This is fine. As long as we have not told anyone that we are
		// commited.
		log.writeMark(LogRecords.PreCommit.toString(), dm.transactionID + " " + str);
		dm.preCommit();
		
		// At this point we know that DMs and DMCs log both says that we are ready to commit. So we can share the good
		// news with others.
		scheduleStatusCheck(coordinatorID, peers, 8);		
		return commit;
	}
	
	/** Commits active transaction. This can be called only after voteRequest returned true for this client.
	 * @param clientID Client making a call - provided by RPC Layer
	 */
	public void commit(int clientID) throws IOException
	{
	    // It is possible that this transaction has been updated already. See not in setTransactionStatus.
	    // Unfortunately in such case we do not know anything about transaction and we can't reconstruct trasnactionID.
	    // So we can't check if this is the same status or not.
	    if (!transactions.containsKey(clientID))
	    {
	        System.out.println("DMC: Ignoring commit/abort request from the client as we think transaction is done. " +
	        		"ClientID: " + clientID);
	        return;
	    }
	        
		DataManager dm = transactions.get(clientID);		
		assert dm.getSatus() == TransactionStatus.PreCommit: "Data Manager should be in PreCommit state for us to" +
				"commit transaction";

        // Write to log. It is possible for the crash to happen after the DM finishes its log updated and deletes the
        // file but before we know that. Restarting would make DMC try to restart DM, which will not be happy.
        log.writeStart(LogRecords.Commited.toString(), dm.transactionID);
        dm.commit();
        log.writeEnd(LogRecords.Commited.toString());
		
		// updated our internal statuses
		transactions.remove(clientID);
		completed.put(dm.transactionID, dm.getSatus());
	}
	
	/** Aborts currently active transaction. Transaction cannot be committed already. No requirement to have called
	 * voteRequest before.
	 * 
	 * @param clientID Client making a call - provided by RPC Layer
	 */
	public void abort(int clientID) throws IOException
	{
	    // It is possible that this transaction has been updated already. See not in setTransactionStatus.
        // Unfortunately in such case we do not know anything about transaction and we can't reconstruct trasnactionID.
        // So we can't check if this is the same status or not.
        if (!transactions.containsKey(clientID))
        {
            System.out.println("DMC: Ignoring commit/abort request from the client as we think transaction is done. " +
                    "ClientID: " + clientID);
            return;
        }

		DataManager dm = transactions.get(clientID);
		assert dm.getSatus() == TransactionStatus.Running || dm.getSatus() == TransactionStatus.PreCommit: 
		    "Data Manager should be in Running or PreCommit state for us to abort transaction";

		// Write to log. It is possible for the crash to happen after the DM finishes its log updated and deletes the
		// file but before we know that. Restarting would make DMC try to restart DM, which will not be happy.
		log.writeStart(LogRecords.Aborted.toString(), dm.transactionID);
		dm.abort();
		log.writeEnd(LogRecords.Aborted.toString());
		
		// updated our internal statuses
		transactions.remove(clientID);
		completed.put(dm.transactionID, dm.getSatus());
	}
	
	/** Returns status of given transaction. This method is exposed via RPC. This method should know about transaction
	 * that it is being asked for.
	 * 
	 * @param clientID ClientID set by RPC and ignored by this function
	 * @param transactionID Transaction to check status of.
	 * @return Transaction status.
	 */
	public TransactionStatus getStatus(int clientID, String transactionID) throws TransactionNotActive
	{
		// Well client here is not necessarily the one that started transaction so we cannot use it to find transaction.
		// We need to find it base don trasnactionID.
		
		// see if we have completed it
		if (completed.containsKey(transactionID))
			return completed.get(transactionID);
		
		// No we do not. Look up in active transactions
		// yeah, could maintain another structure that maps from transcationID to dm or something like that. But this
		// would add significant complication to the code and we do not expect to have a lot of running transactions.
		for (DataManager dm: transactions.values())			
			if (dm.transactionID().equals(transactionID))
				return dm.getSatus();
		
		// It is possible that in some cases client may ask us about transaction that we do not know anything about. This
		// usually means that we are coordinator and TM has not informed us about decision on what to do. However,
		// this can happen only in case when TM sent voteRequest message, but crashed before it made decision. So we
		// tell that we have no idea what the transaction is. TM can still go up and look at trasnaction and make
		// decision. It is just that we need to wait.
		throw new TransactionNotActive();
	}
	
	/** Tries to check status of transaction using peers. As this is call back this one needs to be public. That does
	 * not mean that everyone should start calling it.
	 * 
	 * @param coordinatorID COordinator of this transaction.
	 * @param peers Peers not including coordinator.
	 */
	public void checkStatus(final int coordinatorID, final int[] peers)
	{
		// This can be called if we are restoring transaction or if we are timed out waiting for answer. In any case
		// we need to figure out the status.
		DataManager dm = transactions.get(coordinatorID);
		
		// See maybe someone else figured out the status of transaction. DM will be null only if transaction is done 
		if (dm == null)
			return;
		
		assert dm.getSatus() == TransactionStatus.PreCommit: "We should not try to figure out status of transaction" +
				"that is not in pre-commit";
		
		// Check if RPC Clients are not busy. It is possible that they are working on other transactions. This loop
		// may add non trivial delays in the transaction processing if multiple ones are running. But it should always
		// terminate at some point.
		for (int i: peers)
			if (node.getRPCClient(i).callPending())
			{
				// I guess better luck next time
				scheduleStatusCheck(coordinatorID, peers, 1);
				return;
			}
		
		// We need to check for coordinator client as well
		if (node.getRPCClient(coordinatorID).callPending())
        {
          scheduleStatusCheck(coordinatorID, peers, 1);
          return;
        }
				
		// Spam everyone in the transaction and see what they have.
		ResultGatherer gatherer = new ResultGatherer();
		
		// Lets first ask the coordinator about the status of the transaction
		node.getRPCClient(coordinatorID).invoke("getStatus", new Serializable [] { dm.transactionID },
		                            gatherer.createGatherer());
		                    		
		for (int i: peers)
			// we probably do not want to send this call to ourselves and coordinator again in case it is peer. That
		    // guy has its own preferences
			if (i != node.addr && i != coordinatorID)
				node.getRPCClient(i).invoke("getStatus", new Serializable [] { dm.transactionID },
						gatherer.createGatherer());
		
		// Hey RPC calls are out. Time to wait for their result.
		gatherer.waitAll(new IResultCallback()
		{
			@Override
			@SuppressWarnings("unchecked")
			public void resultCallback(Status status, Serializable result)
			{
				updateTransactionStatus(coordinatorID, peers, (List<Result>) result);
			}
		});
	}
	
	/** Tries to update status based on peer results.
	 * 
	 * @param coordinatorID ID of the client that is coordinating this transaction.
	 * @param peers List of peers participating in the transaction. Does not induce coordinator
	 * @param results List of results returned from the peers. Including coordinator
	 */
	protected void updateTransactionStatus(final int coordinatorID, final int[] peers, 
			List<ResultGatherer.Result> results)
	{
	    // See if we even care about this transaction. If not ignore it. It is possible that it was resolved in some
	    // other way.
	    if (!transactions.containsKey(coordinatorID))
	        return;   
		
		// First see if someone has a result. If not, try later again
        Boolean commit = null;
		for (ResultGatherer.Result dmResult: results)
			// Basically nodes that return three values: PreCommint - they are clues as we are
			// Committed and Aborted - these are the ones that we care
			if (dmResult.status == Status.Success)
			{
                TransactionStatus dmStatus = (TransactionStatus) dmResult.result;
			    // It is possible that we received voteCommit, but others has not. So we need to make sure that we skip
			    // those that are still running or are still pre-commit
                if (dmStatus == TransactionStatus.Running || dmStatus == TransactionStatus.PreCommit)
                    continue;
                
				assert dmStatus == TransactionStatus.Comitted || dmStatus == TransactionStatus.Aborted: "Incorrect " +
						"status returned from node. Received: " + dmStatus;
				
				if (dmStatus == TransactionStatus.Comitted || dmStatus == TransactionStatus.Aborted)
				{
					// now we are talking
					assert commit == null ||
							(commit && dmStatus == TransactionStatus.Comitted) ||
							(!commit && dmStatus == TransactionStatus.Aborted) : "It seems that some nodes " +
									"thinks that we should commit, when others Abort";
							
					commit = dmStatus == TransactionStatus.Comitted;
					// we could terminate the loop at this point, but we do not to make sure that we check
					// asserts
				}
			}
		
		// now we should know if we should commit or not
		if (commit != null)
			try {
				if (commit)
					commit(coordinatorID);
				else
					abort(coordinatorID);
			} catch (IOException ex) {
				throw new InternalError("Failed to update lost transaction", ex);
			}
		else
			// As we do not know what happened, try to ping later and see if someone has some more info. Example
			// because they woke up.
			scheduleStatusCheck(coordinatorID, peers, 8);
	}

	
	/** Schedules invocation of checkStatus for sometime in the future.
	 * 
	 * @param clientID Name of the client that have started transaction. and is coordinating it.
	 * @param peers List of peers that participated in the transaction. Does not include clientID.
	 * @param when How much to delay invocation of this transaction.
	 */
	protected void scheduleStatusCheck(int clientID, int[] peers, int when)
	{
		try
		{
			Method method = getClass().getDeclaredMethod("checkStatus", int.class, int[].class);	
			Callback cb = new Callback(method, this, new Object[] { clientID, peers });
			node.addTimeout(cb, when);
			
		} catch (Exception e)
		{
			throw new InternalError("Failed to register callback", e);
		}		
	}
	
	/** Helper class that stores intermediate state when transaction is being loaded from the log file */
	class State
	{
		/** Coordinator ID of the transaction. It is nopt known in all cases */
		public int coordinatorID = -1;
		
		/** List of peers if transaction moved to PreCommit phase. Does not include coordinator */
		public int[] peers = null;
		
		/** Last seen status of the transaction */
		public TransactionStatus status = null;
		
		/** Instantiates the class
		 * 
		 * @param coordinatorID Coordinator ID
		 * @param status Status of the transaction.
		 */
		public State(int coordinatorID, TransactionStatus status)
		{
			this.coordinatorID = coordinatorID;
			this.status = status;
		}
	}
	
	/** Reads log and loads all active transactions. Log is never cleaned up as we never know when some client may ask
	 * us for transaction status.
	 */
	protected void loadFromLog() throws IOException
	{
		// We need to load from log in two passes. First one is load records in the log and then we would go and see
		// what records needs to be checked how. Second pass also re-creates internal state.
		
		// First pass read log and build temporary state
		// Map from transactionID to status
		Map<String, State> transactions = new HashMap<String, State>();
		for (LogRecord record: log.read())
		{
			assert record.completed || record.action.equals(LogRecords.Commited.toString()) ||
			    record.action.equals(LogRecords.Aborted.toString()) : 
			        "We are using all marks, so all records should be completed, except commit/abort";
		
			// Split params into groups 
			String params[] = record.param.split(" ");
			switch(LogRecords.valueOf(record.action))
			{
				case Started:
					// Create new transaction and write it into the list
					State state = new State(Integer.parseInt(params[0]), TransactionStatus.Running);
					State oldState = transactions.put(generateTransactionID(state.coordinatorID, params[1]), state);
					
					assert params.length == 2: "Incorrect number of parameters";
					assert oldState == null: "This should be first time when we hear about this transaction";
					break;
				case PreCommit:
					// Pre commit means that we may or may not have promised to execute transaction. This is decided
					// by the state of DM. if DM is in PreCommint - then we promised to commit. If it is in state
					// aborted, then we crashed before DM was able to record (and executed) pre-commint. So we have not
					// promised anyone. In fact we failed to send the message. Thus we leave our state as running
					// and deal with this later
					state = transactions.get(params[0]);
					
					// params [1],[2]... are peers.
					state.peers = new int[(params.length - 1)];		
					for (int i = 0; i < params.length - 1; i++)
					  state.peers[i] = Integer.parseInt(params[i+1]);
					break;
				case Commited:
				case Aborted:
				    // Transaction may have been finished or not depending where the crash happened. If this record is
				    // completed then transaction is done if it is not, then we need to see if DM has a log for this
				    // transaction. However, it is just easier to check for the log for both cases.
					TransactionStatus status = record.action.equals(LogRecords.Commited.toString()) ?
					        TransactionStatus.Comitted : TransactionStatus.Aborted;
					
					// So, if no log files, use this method, otherwise make sure that we leave transaction in old state.
					// The only time when this would be the first time that we see transaction is if it was recorded via
					// setTransactionStatus. In that case it would not have log file.
					if (!DataManager.logFileExists(new NodeWrapper(node), params[0]))					
    					if (!transactions.containsKey(params[0]))
    						transactions.put(params[0], new State(-1, status));
    					else
    						transactions.get(params[0]).status = status;
					else
					    assert !record.completed: "If DM has log file, then we should not have completed this record";
					break;
				default:
					assert false: "Unknown state read in the log: " + record.action;
					break;
			}
		}
		
		// Now second phase. See what transaction needs to be fixed up, and update out local structures
		for (Entry<String, State> entry: transactions.entrySet())
			switch (entry.getValue().status)
			{
				case Running:   // Including Pre-Commit
					restoreActiveTransaction(entry.getKey(), entry.getValue());
					break;
				case Comitted:
				case Aborted:
					completed.put(entry.getKey(), entry.getValue().status);
					break;
				default:
					assert false: "We should not be seeing transactions in this state: " + entry.getValue();
					break;
			}
	}

	/** Restores one active transaction based on state collected so far.
	 * 
	 * @param tarnsactionID ID of the transaction
	 * @param transactionState State that transaction is thought to be in. In some cases may not be correct.
	 */
	protected void restoreActiveTransaction(String tarnsactionID, State transactionState) throws IOException
	{
		// Let's see what DM of this transaction thinks
		DataManager dm = new DataManager(tarnsactionID, new NodeWrapper(node), locks);
		switch(dm.getSatus())
		{
		    case Running:
		        // It is possible to have transaction in running state. This means that DMC was able to write
		        // information about transaction start, but DM never started. In such case we just Ignore such
		        // Transaction
		        assert !DataManager.logFileExists(new NodeWrapper(node), tarnsactionID): "Transaction should not be " +
		        		"running";
		        break;
			case PreCommit:
				// Oh, not good. We need to figure out what to do now. We need to ping our peers to learn that
				// It is also alive transaction that we need to remember
				DataManager oldDM = this.transactions.put(transactionState.coordinatorID, dm);
				checkStatus(transactionState.coordinatorID, transactionState.peers);
				assert oldDM == null: "It should be only one active transaction per client ID";
				break;
			case Comitted:
			case Aborted:
				// easy cases - we just updated our state.
				completed.put(dm.transactionID, dm.getSatus());
				log.writeMark(dm.getSatus() == TransactionStatus.Comitted ?  LogRecords.Commited.toString() :
					LogRecords.Aborted.toString(), dm.transactionID);
				break;
			default:
				// If we hit any of these states DM and DMC logs are out of sync
				assert false: "DM should not be reporting this transactions state: " + dm.getSatus();
				break;
		}
	}
	
	/** Verifies that client has an active transaction. if not throws exception.
	 * @param clientID Client ID to check. 
	 */
	protected void transactionExistsGuard(int clientID) throws IOException
	{
		if (!transactions.containsKey(clientID))
		{
		    throw new TransactionNotActive();
		}
	}

	/* Proxy methods for IServerFileIO */
	public void createFile(int clientID, String filename) throws IOException
	{
		transactionExistsGuard(clientID);
		transactions.get(clientID).createFile(filename);
	}

	public String getFileContents(int clientID, String filename) throws IOException
	{
		transactionExistsGuard(clientID);
		return transactions.get(clientID).getFileContents(filename);
	}

	public void putContentsToFile(int clientID, String filename, String contents)
			throws IOException
	{
		transactionExistsGuard(clientID);
		transactions.get(clientID).putContentsToFile(filename, contents);
	}

	public void appendContentsToFile(int clientID, String filename, String contents) throws IOException
	{
		transactionExistsGuard(clientID);
		transactions.get(clientID).appendContentsToFile(filename, contents);
	}

	public boolean deleteFile(int clientID, String filename) throws IOException
	{
		transactionExistsGuard(clientID);
		return transactions.get(clientID).deleteFile(filename);
	}
}
